{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# HBMQTT\n",
    "\n",
    "## 简介\n",
    "\n",
    "开源的Mqtt客户端和broker，使用python标准异步库`asyncio`实现，提供协程并发程序开发的直接API。\n",
    "\n",
    "- GitHub：https://github.com/beerfactory/hbmqtt\n",
    "- 文档：https://hbmqtt.readthedocs.io/en/latest/\n",
    "\n",
    "## Feature\n",
    "\n",
    "HBMQTT实现了完整的MQTT 3.1.1协议规范，并提供了以下特性：\n",
    "\n",
    "- Support QoS 0, QoS 1 and QoS 2 messages flow\n",
    "- 客户端断线自动重连\n",
    "- 通过密钥文件认证（其他方法可通过插件系统实现）\n",
    "- 基础的`$SYS` topics 支持\n",
    "- TCP 和 websocket 支持\n",
    "- TCP 和 websocket 的 SSL 支持\n",
    "- 插件系统\n",
    "\n",
    "## 依赖\n",
    "\n",
    "要求Python >=3.4.3\n",
    "\n",
    "## 安装\n",
    "\n",
    ">pip install hbmqtt==0.9.6\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# 客户端API用法\n",
    "\n",
    "`MQTTClient` 类实现了MQTT协议，用于订阅、发布消息，通过普通TCP或者websocket协议，也可选择安全或者不安全的方式。\n",
    "\n",
    "## 1. 订阅并接收\n",
    "\n",
    "订阅并且打印每条从broker接收到的消息\n",
    "\n",
    "```py\n",
    "import asyncio\n",
    "import logging\n",
    "\n",
    "from hbmqtt.client import MQTTClient, ClientException\n",
    "from hbmqtt.mqtt.constants import QOS_1, QOS_2\n",
    "\n",
    "logger = logging.getLogger('test_sub')\n",
    "\n",
    "my_config = {\n",
    "    'keep_alive': 10,\n",
    "    'ping_delay': 1,\n",
    "    'default_qos': 0,\n",
    "    'default_retain': False,\n",
    "    'auto_reconnect': True,  # 开启自动重连\n",
    "    'reconnect_max_interval': 90,  # 最大重试间隔 秒\n",
    "    'reconnect_retries': -1,  # 一直重试，直到连上\n",
    "}\n",
    "\n",
    "\n",
    "async def uptime_coro():\n",
    "    # 创建客户端实例\n",
    "    C = MQTTClient(client_id='test1', config=my_config, loop=None)\n",
    "    await C.connect('mqtt://test.mosquitto.org/',\n",
    "                    cleansession=True,\n",
    "                    cafile=None,\n",
    "                    extra_headers={})\n",
    "    # 订阅 $SYS/broker/uptime'  QOS=1\n",
    "    # 订阅 $SYS/broker/load/#'  QOS=2\n",
    "    await C.subscribe([\n",
    "        ('$SYS/broker/uptime', QOS_1),\n",
    "        ('$SYS/broker/load/#', QOS_2),\n",
    "    ])\n",
    "    try:\n",
    "        # 接收100条消息\n",
    "        for i in range(0, 100):\n",
    "            # 等待接收消息, 超时时间为None表示一直等待，返回message对象\n",
    "            message = await C.deliver_message(timeout=None)\n",
    "            packet = message.publish_packet\n",
    "            print(\"%d:  %s => %s\" % (i, packet.variable_header.topic_name,\n",
    "                                     str(packet.payload.data)))\n",
    "        # 取消订阅\n",
    "        await C.unsubscribe(['$SYS/broker/uptime', '$SYS/broker/load/#'])\n",
    "        # 客户端主动断开连接\n",
    "        await C.disconnect()\n",
    "    except ClientException as ce:\n",
    "        logger.error(\"Client exception: %s\" % ce)\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    formatter = \"[%(asctime)s] %(name)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s\"\n",
    "    logging.basicConfig(level=logging.DEBUG, format=formatter)\n",
    "    asyncio.get_event_loop().run_until_complete(uptime_coro())\n",
    "\n",
    "\n",
    "```    "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 2. 发布消息\n",
    "\n",
    "异步发布3条消息，每条消息的服务质量都不同。\n",
    "\n",
    "```py\n",
    "import logging\n",
    "import asyncio\n",
    "\n",
    "from hbmqtt.client import MQTTClient, ConnectException\n",
    "from hbmqtt.mqtt.constants import QOS_0, QOS_1, QOS_2\n",
    "\n",
    "logger = logging.getLogger('test_pub')\n",
    "\n",
    "\n",
    "async def test_coro():\n",
    "    C = MQTTClient()\n",
    "    await C.connect('mqtt://test.mosquitto.org/')\n",
    "    tasks = [\n",
    "        asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_0')),\n",
    "        asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_1', qos=QOS_1)),\n",
    "        asyncio.ensure_future(C.publish('a/b', b'TEST MESSAGE WITH QOS_2', qos=QOS_2)),\n",
    "    ]\n",
    "    await asyncio.wait(tasks)\n",
    "    logger.info(\"test_coro messages published\")\n",
    "    await C.disconnect()\n",
    "\n",
    "\n",
    "async def test_coro2():\n",
    "    try:\n",
    "        C = MQTTClient()\n",
    "        ret = await C.connect('mqtt://test.mosquitto.org:1883/')\n",
    "        message = await C.publish('a/b', b'TEST MESSAGE WITH QOS_0', qos=QOS_0)\n",
    "        message = await C.publish('a/b', b'TEST MESSAGE WITH QOS_1', qos=QOS_1)\n",
    "        message = await C.publish('a/b', b'TEST MESSAGE WITH QOS_2', qos=QOS_2)\n",
    "        # print(message)\n",
    "        logger.info(\"test_coro2 messages published\")\n",
    "        await C.disconnect()\n",
    "    except ConnectException as ce:\n",
    "        logger.error(\"Connection failed: %s\" % ce)\n",
    "        asyncio.get_event_loop().stop()\n",
    "\n",
    "\n",
    "if __name__ == '__main__':\n",
    "    formatter = \"[%(asctime)s] %(name)s {%(filename)s:%(lineno)d} %(levelname)s - %(message)s\"\n",
    "    logging.basicConfig(level=logging.DEBUG, format=formatter)\n",
    "    asyncio.get_event_loop().run_until_complete(test_coro())\n",
    "    asyncio.get_event_loop().run_until_complete(test_coro2())\n",
    "\n",
    "```\n",
    "实际上这个脚本执行发布代码通过异步事件循环，先执行完 `test_coro()`函数，后执行 `test_coro2()`函数。两者完成的同样的任务。`test_coro2()` 是按先后顺序发送了3条，而`test_coro()` 是异步发送了3条。通过日志可发现发送的MQTT消息顺序的区别。\n",
    "\n",
    "详细日志信息请自己运行查看。\n",
    "\n",
    "`test_coro()` **异步发布消息的效率更高一些**，（官方文档错了，还是我错了？😱）"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 3. api参数说明\n",
    "\n",
    "### 1. 构造参数  [[source]](https://hbmqtt.readthedocs.io/en/latest/_modules/hbmqtt/client.html#MQTTClient)\n",
    "\n",
    "用于创建客户端实例。\n",
    "\n",
    "```py\n",
    "class hbmqtt.client.MQTTClient(client_id=None, config=None, loop=None)\n",
    "```\n",
    "参数：\n",
    "\n",
    "- `client_id`: 客户端id，如果是None，则通过`hbmqtt.utils.gen_client_id()` 生成随机id\n",
    "- `config`: 客户端配置参数，python字典类型，默认为：\n",
    "\n",
    "    ```py\n",
    "    {\n",
    "        'keep_alive': 10,\n",
    "        'ping_delay': 1,\n",
    "        'default_qos': 0,\n",
    "        'default_retain': False,\n",
    "        'auto_reconnect': True,\n",
    "        'reconnect_max_interval': 10,\n",
    "        'reconnect_retries': 2,\n",
    "    }\n",
    "    ```\n",
    "    **详细配置说明，请看下文**\n",
    "    \n",
    "- `loop`: `asynio`模块的`loop`对象\n",
    "\n",
    "### 2. MQTTClient.connect\n",
    "\n",
    "该方法为协程，用于连接到broker\n",
    "\n",
    "```py\n",
    "client.connect(uri=None, cleansession=None, cafile=None, capath=None, cadata=None, extra_headers={})\n",
    "```\n",
    "参数：\n",
    "\n",
    "- `uri`：Broker 地址，格式见上一篇介绍\n",
    "- `cleansession`：是否清除原有会话\n",
    "- `cafile`： 服务证书文件（可选，用于安全连接）\n",
    "- `capath`： 服务证书路径（可选，用于安全连接）\n",
    "- `cadata`： 服务证书数据（可选，用于安全连接）\n",
    "- `extra_headers`：用于websocket初始化连接附加的http请求头\n",
    "\n",
    "返回值：\n",
    "\n",
    "   [CONNACK](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718033) 消息返回值\n",
    "   \n",
    "异常：\n",
    "\n",
    "  连接失败抛出`hbmqtt.client.ConnectException`\n",
    "\n",
    "### 3. MQTTClient.disconnect\n",
    "\n",
    "协程，断开连接\n",
    "\n",
    "```py\n",
    "client.disconnect()\n",
    "```\n",
    "\n",
    "### 4. MQTTClient.reconnect\n",
    "\n",
    "协程，重连功能。重试次数和间隔可以通过`reconnect_max_interval` 和 `reconnect_retries` 参数控制\n",
    "\n",
    "```py\n",
    "client.reconnect(cleansession=None)\n",
    "```\n",
    "\n",
    "参数\n",
    "\n",
    "- `cleansession`：是否清除以前会话\n",
    "\n",
    "返回值：\n",
    "\n",
    "   [CONNACK](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718033) 消息返回值\n",
    "   \n",
    "异常：\n",
    "\n",
    "  连接失败抛出`hbmqtt.client.ConnectException`\n",
    "  \n",
    "### 5. MQTTClient.ping\n",
    "\n",
    "协程，发送 ping 消息\n",
    "\n",
    "```py\n",
    "client.ping()\n",
    "```\n",
    "\n",
    "### 6. MQTTClient.publish\n",
    "\n",
    "协程，发布消息。并根据服务质量等待确认消息\n",
    "\n",
    "```py\n",
    "client.publish(topic, message, qos=None, retain=None)\n",
    "```\n",
    "\n",
    "参数：\n",
    "\n",
    "- `topic`： 主题\n",
    "- `message`：消息载荷，bytes类型\n",
    "- `qos`：服务质量，取值0, 1, 2. 默认值根据连接配置中的 `default_qos` 决定\n",
    "- `retain`：retail 标志位，默认值根据连接配置中的`default_retain` 决定\n",
    "\n",
    "### 7. MQTTClient.subscribe\n",
    "\n",
    "协程，订阅topic，发送订阅消息，并等待broker的确认包。\n",
    "\n",
    "```py\n",
    "client.subscribe(topics)\n",
    "```\n",
    "\n",
    "参数：\n",
    "\n",
    "- `topics`: 主题和QoS 的元组 组成的列表，如\n",
    "```py\n",
    "  [\n",
    "    ('$SYS/broker/uptime', QOS_1),\n",
    "    ('$SYS/broker/load/#', QOS_2)\n",
    "   ]\n",
    "```\n",
    "\n",
    "返回：\n",
    "\n",
    "[SUBACK](http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html#_Toc398718068) 确认消息返回值\n",
    "\n",
    "### 8. MQTTClient.unsubscribe\n",
    "\n",
    "协程，取消订阅。\n",
    "\n",
    "```py\n",
    "client.unsubscribe(topics)\n",
    "```\n",
    "\n",
    "参数：\n",
    "\n",
    "- topics：主题列表，如：\n",
    "```py\n",
    "['$SYS/broker/uptime', '$SYS/broker/load/#']\n",
    "```\n",
    "\n",
    "### 9. MQTTClient.deliver_message\n",
    "\n",
    "协程，等待从broker 接收下一条消息，如果没有收到消息，则会一直等待，或者指定等待的超时时间 `timeout`。\n",
    "\n",
    "```py\n",
    "client.deliver_message(timeout=None)\n",
    "```\n",
    "\n",
    "参数：\n",
    "\n",
    "- timeout: 等待消息到达时间，单位：秒。`None`：一直等待\n",
    "\n",
    "返回：\n",
    "\n",
    "`hbmqtt.session.ApplicationMessage` 实例，包含消息的所有属性。\n",
    "\n",
    "异常：\n",
    "\n",
    "如果timeout参数超时，抛出 `asyncio.TimeoutError`。"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## 4. 连接配置参数\n",
    "\n",
    "`MQTTClient.__init__` 接收一个 `config` 参数，python的字典类型，包含以下字段：\n",
    "\n",
    "- `keep_alive`：keep alive时间，单位秒（默认10秒），客户端会自动发送 ping消息给broker，以维持客户端不掉线。\n",
    "- `ping_delay` 在keep alive 超时之前自动发送 ping 消息的延迟，默认 1秒 \n",
    "- `default_qos`：默认消息质量，**在publish() 方法未指定qos参数时起作用**。\n",
    "- `default_retain`：默认 retain值（False），在publish() 方法未指定qos参数时起作用。\n",
    "- `auto_reconnect`：是否自动重连，默认 `True`\n",
    "- `reconnect_max_interval`: 两次重连之间的最大间隔，默认为10s\n",
    "- `reconnect_retries`：最大重试次数，默认2次\n",
    "\n",
    "默认QoS等级和retain值，可以被额外的`topic` 参数重写，例如\n",
    "\n",
    "```py\n",
    "config = {\n",
    "    'keep_alive': 10,\n",
    "    'ping_delay': 1,\n",
    "    'default_qos': 0,\n",
    "    'default_retain': False,\n",
    "    'auto_reconnect': True,\n",
    "    'reconnect_max_interval': 5,\n",
    "    'reconnect_retries': 10,\n",
    "    'topics': {\n",
    "        '/test': { 'qos': 1 },\n",
    "        '/some_topic': { 'qos': 2, 'retain': True }\n",
    "    }\n",
    "}\n",
    "```\n",
    "\n",
    "以上配置会设置所有消的默认QoS为0，retain值为False，除了：\n",
    "\n",
    "- 发送到topic 为 `/test` 的消息，QoS默认值为1\n",
    "- 发送到topic 为 `/some_topic`的消息,QoS默认值为2, retain值为True\n"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.3"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {},
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "183.545px"
   },
   "toc_section_display": true,
   "toc_window_display": true
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}
